﻿using System;
using System.Collections.Generic;
using System.Text;
using Common.Logging;
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Rmq.Center
{
    public class BaseConsumer<T> : IDisposable
    {
        private static readonly ILog Logger = LogManager.GetLogger(typeof(BaseConsumer<T>));

        private readonly IConnection _connection;
        private readonly IModel _channel;

        private readonly string _queueName;
        private readonly string _routingKey;

        public BaseConsumer(string queueName, string routingKey)
        {
            var factory = new ConnectionFactory()
            {
                HostName = Const.HostName,
                UserName = Const.UserName,
                Password = Const.Password,
            };
            factory.AutomaticRecoveryEnabled = true;
            factory.TopologyRecoveryEnabled = true;
            _connection = factory.CreateConnection();
            _channel = _connection.CreateModel();

            _queueName = queueName;
            _routingKey = routingKey;
        }

        public virtual void Subscribe(ConsumerDelegate<T> received)
        {
            // declare normal exchange
            _channel.ExchangeDeclare(
                exchange: Const.ExchangeName,
                type: "topic",
                durable: true,
                autoDelete: false,
                arguments: null);

            // declare delayed exchange
            var exchangeArguments = new Dictionary<string, object>();
            exchangeArguments.Add("x-delayed-type", "topic");
            _channel.ExchangeDeclare(
                exchange: Const.DelayedExchangeName,
                type: "x-delayed-message",
                durable: true,
                autoDelete: false,
                arguments: exchangeArguments);

            _channel.QueueDeclare(queue: _queueName,
                                 durable: true,
                                 exclusive: false,
                                 autoDelete: false,
                                 arguments: null);

            // declare normal exchange queue bind
            _channel.QueueBind(queue: _queueName,
                              exchange: Const.ExchangeName,
                              routingKey: _routingKey);

            // declare delayed exchange queue bind
            _channel.QueueBind(queue: _queueName,
                              exchange: Const.DelayedExchangeName,
                              routingKey: _routingKey);

            _channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);

            var consumer = new EventingBasicConsumer(_channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body);
                var routingKey = ea.RoutingKey;
                
                var wrappedMsg = JsonConvert.DeserializeObject<Msg<T>>(message);
                wrappedMsg.ReceivedCount++;

                var result = new ConsumedResult();
                try
                {
                    if (received != null)
                        result = received(wrappedMsg.MsgId, wrappedMsg.ReceivedCount, wrappedMsg.MsgProduceDt,
                            wrappedMsg.OrgMsg, new WrapperChannel(_channel));
                }
                catch (Exception ex)
                {
                   
                    Logger.ErrorFormat("消息队列消费出错 消息内容：{0}, 错误信息： {1}", message, JsonConvert.SerializeObject(ex));
                    result.Ack = true;
                    // 如果重试次数大于10 且出现异常的话，则ack掉本消息
                    //if (wrappedMsg.ReceivedCount >= 10)
                    //{
                    //    result.Ack = true;
                    //}
                }

                if (result.Ack)
                {
                    _channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                }
                else
                {
                    var reBody = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(wrappedMsg));

                    if (result.Delay > 0)
                    {
                        var properties = _channel.CreateBasicProperties();
                        properties.Persistent = true;
                        properties.Headers = new Dictionary<string, object>();
                        properties.Headers.Add("x-delay", result.Delay);

                        _channel.BasicPublish(exchange: Const.DelayedExchangeName,
                            routingKey: _routingKey,
                            basicProperties: properties,
                            body: reBody);
                    }
                    else
                    {
                        var properties = _channel.CreateBasicProperties();
                        properties.Persistent = true;

                        _channel.BasicPublish(exchange: Const.ExchangeName,
                            routingKey: _routingKey,
                            basicProperties: properties,
                            body: reBody);
                    }

                    _channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                }
            };

            _channel.BasicConsume(queue: _queueName,
                                 noAck: false,
                                 consumer: consumer);
        }

        public virtual void Dispose()
        {
            _channel.Dispose();
            _connection.Dispose();
        }
    }
}